import com.zuikaku.pojo.Order;

import com.zuikaku.sink.MyRedisSink;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import java.math.BigDecimal;
import java.util.Arrays;
import java.util.UUID;

public class RedisSinkDemo {
    public static void main(String[] args) {
        //创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment();
        environment.setParallelism(1);
        //根据order获取source
        DataStreamSource<Order> orderDS = environment.fromElements(new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1001,new BigDecimal("6999.99")),
                new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1002,new BigDecimal("6979.99")),
                new Order(UUID.randomUUID().toString(),"iPhone 15 Pro Max",1001,new BigDecimal("8999.99")),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1004,new BigDecimal("1999.99")),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1005,new BigDecimal("1969.99")));
        //order source转化为要处理到redis的Tuple2<String,Integer>，源数据进来已经是一个一个的对象，所用用map，而不是flatmap
        DataStream<Tuple2<String,Integer>> mapDS = orderDS.map(new MapFunction<Order, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String,Integer> map(Order value) throws Exception {
                return new Tuple2<>(value.getItemName(),1);//订单商品数量默认为1
            }
        });
        //对mapDS进行分组，以title相同的为一组
        KeyedStream<Tuple2<String,Integer>,String> keyedStream = mapDS.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;//基于元组中第一个元素作为key，以该字段作为分组条件
            }
        });
        //进行求和,指定需要求和字段的下标，也就是Tuple<String,Integer>的integer
        DataStream<Tuple2<String,Integer>> sumed = keyedStream.sum(1);
        sumed.print("根据itemName分组求和后");

        //输出到redis
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setPort(6379).build();
        sumed.addSink(new RedisSink<>(conf,new MyRedisSink()));



        try {
            environment.execute("key by & sum job");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
